package com.yuan.kafkastudy.producer;

import com.yuan.kafkastudy.constant.MsgType;
import com.yuan.kafkastudy.constant.TopicName;
import com.yuan.kafkastudy.proto.PlatformProto;
import com.yuan.kafkastudy.proto.protoJava.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

public class KafkaProducer1 {

    public static void main(String[] args) {
        //创建配置
        Map<String, Object> config = new HashMap();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());

        //创建生产者
        KafkaProducer<String, byte[]> producer = new KafkaProducer(config);
        //创建数据
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < 1010000; i++) {
            SimProto.Simulation.Builder simPro = SimProto.Simulation.newBuilder();
            simPro.setSimTimeVal(5);
            simPro.setSimTimeUnit(SimProto.TimeUnit.MINUTE);
            simPro.setRandomSeed(1);
            byte[] simBytes = simPro.build().toByteArray();
            send(simBytes, "1", producer);

            PlatformProto.Platform.Builder plarformPro = PlatformProto.Platform.newBuilder();
            plarformPro.setName("B-1B");
            plarformPro.setType("WSF_AIR_PLATFORM");
            byte[] platformBytes = plarformPro.build().toByteArray();
            send(platformBytes, "2", producer);

            SensorProto.Sensor.Builder sensorPro = SensorProto.Sensor.newBuilder();
            sensorPro.setName("radar");
            sensorPro.setType("WSF_RADAR_SENSOR");
            sensorPro.setParam("{,}");
            byte[] sensorBytes = sensorPro.build().toByteArray();
            send(sensorBytes, "3", producer);

            CommProto.Comm.Builder commPro = CommProto.Comm.newBuilder();
            commPro.setName("UHF_COMM");
            commPro.setType("WSF_COMM");
            commPro.setParam("{,}");
            byte[] commBytes = commPro.build().toByteArray();
            send(commBytes, "4", producer);

            WeaponProto.Weapon.Builder weaponPro = WeaponProto.Weapon.newBuilder();
            weaponPro.setName("DF-1");
            weaponPro.setType("AAM");
            weaponPro.setParam("{,}");
            byte[] weaponBytes = weaponPro.build().toByteArray();
            send(weaponBytes, "5", producer);

            EnvironmentProto.Environment.Builder environmentPro = EnvironmentProto.Environment.newBuilder();
            environmentPro.setName("environmnent");
            environmentPro.setParam("{,}");
            byte[] envirBytes = environmentPro.build().toByteArray();
            send(envirBytes, "6", producer);
        }
        long endTime = System.currentTimeMillis();
        System.out.println("耗时:"+(endTime - startTime));
        //关闭生产者
        producer.close();
    }
    private static void send(byte[] message, String key , Producer producer){
        ByteBuffer buf = ByteBuffer.allocate(1 + message.length);
        if (key.equals("1")){
            buf.put(MsgType.SIMULATION);          // 第 0 字节：类型
        }
        if (key.equals("2")){
            buf.put(MsgType.PLATFORM);
        }
        if (key.equals("3")){
            buf.put(MsgType.SENSOR);
        }
        if (key.equals("4")){
            buf.put(MsgType.COMM);
        }
        if (key.equals("5")){
            buf.put(MsgType.WEAPON);
        }
        if (key.equals("6")){
            buf.put(MsgType.ENVIRONMENT);
        }
        buf.put(message);       // 后面：proto 二进制
        ProducerRecord<String, byte[]> record = new ProducerRecord<>(TopicName.TOPIC_NAME_1, key, buf.array());
        producer.send(record);
    }
}
